package com.demo.pulsar;

import com.github.shyiko.mysql.binlog.BinaryLogClient;
import com.github.shyiko.mysql.binlog.event.*;
import com.github.shyiko.mysql.binlog.event.deserialization.EventDeserializer;
import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.client.api.*;

import java.io.IOException;
import java.io.Serializable;
import java.sql.*;
import java.util.Properties;
import java.util.concurrent.TimeUnit;
import java.util.List;
import java.util.Arrays;
import java.util.Map;

@Slf4j
public class PulsarMySQLToPostgreSQLSync {
    private static Properties properties = new Properties();
    private static Connection mysqlConnection;
    private static Connection pgConnection;
    private static PulsarClient pulsarClient;
    private static String mysqlTable;
    private static String pgTable;
    private static String mysqlDatabase;

    public static void main(String[] args) {
        try {
            loadConfig();
            initConnections();
            setupPulsarClient();
            createPostgresTableIfNotExists();
            syncHistoricalData();
            startBinlogSync();
        } catch (Exception e) {
            log.error("同步过程发生错误: ", e);
        } finally {
            cleanup();
        }
    }

    private static void loadConfig() throws IOException {
        properties.load(PulsarMySQLToPostgreSQLSync.class.getClassLoader().getResourceAsStream("application.properties"));
        if (properties.isEmpty()) {
            throw new IOException("无法加载配置文件 application.properties");
        }
        mysqlTable = properties.getProperty("mysql.table");
        pgTable = properties.getProperty("postgresql.table");
        mysqlDatabase = properties.getProperty("mysql.database");
        log.info("成功加载配置文件");
    }

    private static void createPostgresTableIfNotExists() throws SQLException {
        // 先删除旧表
        try (Statement stmt = pgConnection.createStatement()) {
            stmt.execute("DROP TABLE IF EXISTS " + pgTable);
            log.info("已删除旧表（如果存在）");
        }

        // 创建新表
        String createTableSQL = "CREATE TABLE IF NOT EXISTS " + pgTable + " ("
                + "id INT PRIMARY KEY,"
                + "name VARCHAR(255) NOT NULL,"
                + "gender VARCHAR(50) DEFAULT 'UNKNOWN'"
                + ")";
        
        try (Statement stmt = pgConnection.createStatement()) {
            stmt.execute(createTableSQL);
            log.info("GaussDB表创建成功");
        }
    }

    private static void initConnections() throws SQLException {
        // MySQL连接
        String mysqlUrl = String.format("jdbc:mysql://%s:%s/%s?%s",
                properties.getProperty("mysql.host"),
                properties.getProperty("mysql.port"),
                mysqlDatabase,
                properties.getProperty("mysql.url.params"));
        
        mysqlConnection = DriverManager.getConnection(mysqlUrl,
                properties.getProperty("mysql.username"),
                properties.getProperty("mysql.password"));
        log.info("MySQL连接成功");

        // GaussDB连接
        String pgUrl = String.format("jdbc:postgresql://%s:%s/%s",
                properties.getProperty("postgresql.host"),
                properties.getProperty("postgresql.port"),
                properties.getProperty("postgresql.database"));
        
        pgConnection = DriverManager.getConnection(pgUrl,
                properties.getProperty("postgresql.username"),
                properties.getProperty("postgresql.password"));
        log.info("GaussDB连接成功");
    }

    private static void setupPulsarClient() throws PulsarClientException {
        pulsarClient = PulsarClient.builder()
                .serviceUrl(properties.getProperty("pulsar.service.url"))
                .build();
        log.info("Pulsar客户端连接成功");
    }

    private static void syncHistoricalData() throws SQLException, InterruptedException {
        log.info("开始同步历史数据...");
        String selectSQL = String.format("SELECT * FROM %s", mysqlTable);
        try (Statement stmt = mysqlConnection.createStatement();
             ResultSet rs = stmt.executeQuery(selectSQL)) {
            
            while (rs.next()) {
                syncRecord(rs.getInt("id"), rs.getString("name"), rs.getString("gender"));
            }
        }
        log.info("历史数据同步完成");
    }

    private static void startBinlogSync() {
        try {
            log.info("准备连接MySQL binlog...");
            log.info("配置信息: host={}, port={}, user={}, database={}, table={}",
                properties.getProperty("mysql.host"),
                properties.getProperty("mysql.port"),
                properties.getProperty("mysql.username"),
                mysqlDatabase,
                mysqlTable
            );

            BinaryLogClient client = new BinaryLogClient(
                properties.getProperty("mysql.host"),
                Integer.parseInt(properties.getProperty("mysql.port")),
                properties.getProperty("mysql.username"),
                properties.getProperty("mysql.password")
            );

            // 设置服务器ID
            client.setServerId(100);
            
            // 设置心跳间隔（毫秒）
            client.setHeartbeatInterval(TimeUnit.SECONDS.toMillis(5));
            
            // 设置保持连接
            client.setKeepAlive(true);
            
            // 设置连接超时（毫秒）
            client.setConnectTimeout(5000);
            
            // 配置事件反序列化器
            EventDeserializer eventDeserializer = new EventDeserializer();
            eventDeserializer.setCompatibilityMode(
                EventDeserializer.CompatibilityMode.DATE_AND_TIME_AS_LONG,
                EventDeserializer.CompatibilityMode.CHAR_AND_BINARY_AS_BYTE_ARRAY
            );
            client.setEventDeserializer(eventDeserializer);

            // 注册生命周期监听器
            client.registerLifecycleListener(new BinaryLogClient.LifecycleListener() {
                @Override
                public void onConnect(BinaryLogClient client) {
                    log.info("已连接到MySQL binlog服务器");
                    log.info("当前binlog文件: {}, 位置: {}", client.getBinlogFilename(), client.getBinlogPosition());
                    
                    // 连接后打印一些重要信息
                    try {
                        Statement stmt = mysqlConnection.createStatement();
                        ResultSet rs = stmt.executeQuery("SHOW MASTER STATUS");
                        if (rs.next()) {
                            log.info("MySQL主服务器状态: File={}, Position={}, Binlog_Do_DB={}, Binlog_Ignore_DB={}", 
                                rs.getString("File"),
                                rs.getString("Position"),
                                rs.getString("Binlog_Do_DB"),
                                rs.getString("Binlog_Ignore_DB")
                            );
                        }
                        rs.close();
                        
                        rs = stmt.executeQuery("SHOW VARIABLES LIKE 'binlog_format'");
                        if (rs.next()) {
                            log.info("Binlog格式: {}", rs.getString("Value"));
                        }
                        rs.close();
                        stmt.close();
                    } catch (SQLException e) {
                        log.error("获取MySQL状态信息失败: ", e);
                    }
                }

                @Override
                public void onCommunicationFailure(BinaryLogClient client, Exception ex) {
                    log.error("Binlog通信失败: ", ex);
                }

                @Override
                public void onEventDeserializationFailure(BinaryLogClient client, Exception ex) {
                    log.error("Binlog事件解析失败: ", ex);
                }

                @Override
                public void onDisconnect(BinaryLogClient client) {
                    log.info("与MySQL binlog服务器断开连接");
                }
            });

            // 注册事���监听器
            client.registerEventListener(event -> {
                EventHeader header = event.getHeader();
                EventData data = event.getData();
                
                // 打印事件基本信息
                log.info("收到binlog事件: type={}, timestamp={}, serverId={}, binlogFilename={}, position={}, dataLength={}", 
                    header.getEventType(),
                    header.getTimestamp(),
                    header.getServerId(),
                    client.getBinlogFilename(),
                    client.getBinlogPosition(),
                    header.getDataLength()
                );

                // 打印事件数据的详细信息
                if (data != null) {
                    log.info("事件数据: {}", data.toString());
                }

                // 处理表映射事件
                if (data instanceof TableMapEventData) {
                    TableMapEventData tableData = (TableMapEventData) data;
                    log.info("表映射事件: database={}, table={}, tableId={}, columnTypes={}", 
                        tableData.getDatabase(), 
                        tableData.getTable(),
                        tableData.getTableId(),
                        Arrays.toString(tableData.getColumnTypes())
                    );
                    
                    // 只处理目标表的事件
                    if (!tableData.getDatabase().equals(mysqlDatabase) || !tableData.getTable().equals(mysqlTable)) {
                        log.info("跳过非目标表的事件: {}.{}", tableData.getDatabase(), tableData.getTable());
                        return;
                    }
                }

                // 处理DML事件
                if (data instanceof WriteRowsEventData) {
                    WriteRowsEventData eventData = (WriteRowsEventData) data;
                    log.info("INSERT事件: tableId={}, includedColumns={}, rows={}", 
                        eventData.getTableId(), 
                        eventData.getIncludedColumns(),
                        eventData.getRows()
                    );
                    handleWriteRows(eventData);
                } else if (data instanceof UpdateRowsEventData) {
                    UpdateRowsEventData eventData = (UpdateRowsEventData) data;
                    log.info("UPDATE事件: tableId={}, includedColumns={}, rows={}", 
                        eventData.getTableId(), 
                        eventData.getIncludedColumns(),
                        eventData.getRows()
                    );
                    handleUpdateRows(eventData);
                } else if (data instanceof DeleteRowsEventData) {
                    DeleteRowsEventData eventData = (DeleteRowsEventData) data;
                    log.info("DELETE事件: tableId={}, includedColumns={}, rows={}", 
                        eventData.getTableId(), 
                        eventData.getIncludedColumns(),
                        eventData.getRows()
                    );
                    handleDeleteRows(eventData);
                }
            });

            // 启动客户端
            log.info("开始连接MySQL binlog服务器...");
            client.connect();
            log.info("Binlog监听已启动");

        } catch (Exception e) {
            log.error("Binlog监听错误: ", e);
        }
    }

    private static void handleWriteRows(WriteRowsEventData eventData) {
        try {
            log.info("开始处理INSERT事件，行数: {}", eventData.getRows().size());
            log.info("列信息: {}", Arrays.toString(eventData.getIncludedColumns().stream().toArray()));
            
            for (Serializable[] row : eventData.getRows()) {
                log.info("处理行数据，数组长度: {}, 数据: {}", row.length, Arrays.toString(row));
                
                // 安全地获取数据
                if (row.length < 1) {
                    log.error("行数据不完整：缺少id字段");
                    continue;
                }
                int id = (Integer) row[0];
                
                String name = row.length > 1 ? (String) row[1] : "Unknown";
                String gender = row.length > 2 && row[2] != null ? (String) row[2] : "UNKNOWN";
                
                log.info("解析后的数据: id={}, name={}, gender={}", id, name, gender);
                syncRecord(id, name, gender);
            }
        } catch (Exception e) {
            log.error("处理INSERT事件错误: ", e);
            log.error("事件详情: tableId={}, includedColumns={}, rows={}", 
                eventData.getTableId(), 
                eventData.getIncludedColumns(),
                eventData.getRows()
            );
        }
    }

    private static void handleUpdateRows(UpdateRowsEventData eventData) {
        try {
            log.info("开始处理UPDATE事件，行数: {}", eventData.getRows().size());
            log.info("列信息: {}", Arrays.toString(eventData.getIncludedColumns().stream().toArray()));
            
            for (Map.Entry<Serializable[], Serializable[]> row : eventData.getRows()) {
                Serializable[] newRow = row.getValue();
                log.info("处理行数据，数组长度: {}, 数据: {}", newRow.length, Arrays.toString(newRow));
                
                // 安全地获取数据
                if (newRow.length < 1) {
                    log.error("行数据不完整：缺少id字段");
                    continue;
                }
                int id = (Integer) newRow[0];
                
                String name = newRow.length > 1 ? (String) newRow[1] : "Unknown";
                String gender = newRow.length > 2 && newRow[2] != null ? (String) newRow[2] : "UNKNOWN";
                
                log.info("解析后的数据: id={}, name={}, gender={}", id, name, gender);
                syncRecord(id, name, gender);
            }
        } catch (Exception e) {
            log.error("处理UPDATE事件错误: ", e);
            log.error("事件详情: tableId={}, includedColumns={}, rows={}", 
                eventData.getTableId(), 
                eventData.getIncludedColumns(),
                eventData.getRows()
            );
        }
    }

    private static void handleDeleteRows(DeleteRowsEventData eventData) {
        try {
            log.info("开始处理DELETE事件，行数: {}", eventData.getRows().size());
            log.info("列信息: {}", Arrays.toString(eventData.getIncludedColumns().stream().toArray()));
            
            for (Serializable[] row : eventData.getRows()) {
                log.info("处理行数据，数组长度: {}, 数据: {}", row.length, Arrays.toString(row));
                
                // 安全地获取数据
                if (row.length < 1) {
                    log.error("行数据不完整：缺少id字段");
                    continue;
                }
                int id = (Integer) row[0];
                
                log.info("解析后的数据: id={}", id);
                deleteRecord(id);
            }
        } catch (Exception e) {
            log.error("处理DELETE事件错误: ", e);
            log.error("事件详情: tableId={}, includedColumns={}, rows={}", 
                eventData.getTableId(), 
                eventData.getIncludedColumns(),
                eventData.getRows()
            );
        }
    }

    private static void syncRecord(int id, String name, String gender) {
        try {
            // 检查记录是否存在
            String checkSQL = String.format("SELECT COUNT(*) FROM %s WHERE id = ?", pgTable);
            PreparedStatement checkStmt = pgConnection.prepareStatement(checkSQL);
            checkStmt.setInt(1, id);
            ResultSet checkRs = checkStmt.executeQuery();
            checkRs.next();
            boolean exists = checkRs.getInt(1) > 0;
            checkRs.close();
            checkStmt.close();

            if (exists) {
                // 更新记录
                String updateSQL = String.format(
                    "UPDATE %s SET name = ?, gender = ? WHERE id = ?",
                    pgTable
                );
                PreparedStatement updateStmt = pgConnection.prepareStatement(updateSQL);
                updateStmt.setString(1, name);
                updateStmt.setString(2, gender != null ? gender : "UNKNOWN");
                updateStmt.setInt(3, id);
                updateStmt.executeUpdate();
                updateStmt.close();
                log.info("更新记录: ID={}, 姓名={}, 性别={}", id, name, gender);
            } else {
                // 插入记录
                String insertSQL = String.format(
                    "INSERT INTO %s (id, name, gender) VALUES (?, ?, ?)",
                    pgTable
                );
                PreparedStatement insertStmt = pgConnection.prepareStatement(insertSQL);
                insertStmt.setInt(1, id);
                insertStmt.setString(2, name);
                insertStmt.setString(3, gender != null ? gender : "UNKNOWN");
                insertStmt.executeUpdate();
                insertStmt.close();
                log.info("插入记录: ID={}, 姓名={}, 性别={}", id, name, gender);
            }
        } catch (SQLException e) {
            log.error("同步记录错误: ", e);
        }
    }

    private static void deleteRecord(int id) {
        try {
            String deleteSQL = String.format("DELETE FROM %s WHERE id = ?", pgTable);
            PreparedStatement deleteStmt = pgConnection.prepareStatement(deleteSQL);
            deleteStmt.setInt(1, id);
            deleteStmt.executeUpdate();
            deleteStmt.close();
            log.info("删除记录: ID={}", id);
        } catch (SQLException e) {
            log.error("删除记录错误: ", e);
        }
    }

    private static void cleanup() {
        try {
            if (mysqlConnection != null) mysqlConnection.close();
            if (pgConnection != null) pgConnection.close();
            if (pulsarClient != null) pulsarClient.close();
            log.info("清理资源完成");
        } catch (Exception e) {
            log.error("清理资源时发生错误: ", e);
        }
    }
} 